package com.iwhalecloud.uncc.mq.listener;

import com.ctg.mq.api.bean.MQResult;
import com.ctg.mq.api.listener.ConsumerTopicListener;
import com.ctg.mq.api.listener.ConsumerTopicStatus;
import com.iwhalecloud.uncc.utils.ZipUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class MqTopicListener extends MqListener implements MessageListenerConcurrently, ConsumerTopicListener {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        boolean flag = true;
        try {
            if (list != null && !list.isEmpty()) {
//                String message = new String(ZipUtil.uzip(list.get(0).getBody()), StandardCharsets.UTF_8);
                flag = this.handler.dealMessage(list);
            }
        }catch (Exception e){
            log.error("Consume Message Error! Handler is " + this.handler.getClass().getName() + ", errorMsg:" +
                    ExceptionUtils.getStackTrace(e));
            flag = false;
        }
        if(flag){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }else{
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    @Override
    public ConsumerTopicStatus onMessage(List<MQResult> list) {
        boolean flag = true;
        try{
            if(list != null && !list.isEmpty()){
//                String message = new String(ZipUtil.uzip(list.get(0).getMessage().getBody()), StandardCharsets.UTF_8);
                flag = this.handler.dealMessage(list);
            }
        }catch (Exception e){
            log.error("Consume Message Error! Handler is " + this.handler.getClass().getName() + ", errorMsg:" +
                    ExceptionUtils.getStackTrace(e));
            flag = false;
        }
        if(flag){
            return ConsumerTopicStatus.CONSUME_SUCCESS;
        }else{
            return ConsumerTopicStatus.RECONSUME_LATER;
        }
    }

}
